package com.future.WebSocket;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.future.allUtils.constants.RedisCacheConstants;
import com.future.domain.ChatMessage;
import com.future.domain.DTO.ChatMessageDTO;
import com.future.domain.DTO.MessageDTO;
import com.future.domain.DTO.NotificationDTO;
import com.future.domain.vo.NotificationType;
import com.future.mapper.ChatRoomMemberMapper;
import com.future.service.ChatMemberService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
/**
 * 消息处理器
 */
public class ChatWebSocketHandler extends TextWebSocketHandler {
    /**
     * 成功建立连接后触发
     * 初始化用户会话、记录日志、加入在线列表
     *
     * @param session
     * @throws Exception
     */

    @Autowired
    private ChatRoomMemberMapper chatRoomMemberMapper;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private PravateChatstrategy pravateChatstrategy;
    @Autowired
    private GroupChatstrategy groupChatstrategy;
    @Autowired
    private WebSocketSessionManager sessionManager;
    @Autowired
    private ChatMemberService chatService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    @Qualifier("offlineMessageExecutor") // 指定使用离线消息线程池
    private ExecutorService offlineExecutor;
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        // 获取用户id
        Long userId = (Long) session.getAttributes().get("userId");
        if (userId == null) {
            session.close(CloseStatus.POLICY_VIOLATION.withReason("未认证用户"));
            log.info("连接失败，连接用户未认证");
            return;
        }
        sessionManager.addSession(userId, session);
        log.info("连接成功，userId:{}", userId);
        // 用户上线，更新在线状态
        chatService.updateUserOnlineStatus(userId,1);
        //推送离线通知
        offNotification(NotificationType.Like.getDescription(),userId,session);
        offNotification(NotificationType.Comment.getDescription(),userId,session);
        offNotification(NotificationType.Mark.getDescription(),userId,session);
        offNotification(NotificationType.Question.getDescription(),userId,session);
        //推送离线消息
        offMessage(userId,session);
    }
    //处理离线消息，推送离线消息
        public void offNotification(String notificationType,Long userId,WebSocketSession session)
        {
            //从redis中获取离线消息并解析
            String offCommentNotification= RedisCacheConstants.NOTIFICATION_OFFLINE_PREFIX +notificationType+ ":" +userId;
            Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet().reverseRangeWithScores(offCommentNotification, 0, -1);
            List<NotificationDTO> notificationDtos = Optional.ofNullable(typedTuples)
                    .orElse(Collections.emptySet())
                    .stream()
                    .map(typedTuple -> {
                        try {
                            return objectMapper.readValue(typedTuple.getValue(), NotificationDTO.class);
                        } catch (JsonProcessingException e) {
                            log.error("解析zset失败" + e);
                            return null; // 解析失败的通知跳过
                        }
                    })
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            if (notificationDtos.isEmpty()) {
                log.info("用户{}无离线{}通知", userId, notificationType);
                return;
            }
            // 使用线程池异步推送每条通知
            // 为避免并发问题，给session加同步锁
            Object lock = new Object();
            for (NotificationDTO notificationDTO : notificationDtos) {
                // 提交任务到线程池
                offlineExecutor.submit(() -> {
                    try {
                        if (!session.isOpen()) {
                            log.warn("用户{}的WebSocket连接已关闭，无法推送通知", userId);
                            return;
                        }
                        // 同步发送（避免多线程操作session导致异常）
                        synchronized (lock) {
                            String json = objectMapper.writeValueAsString(notificationDTO);
                            session.sendMessage(new TextMessage(json));
                        }
                        log.info("用户{}的离线{}通知推送成功，通知ID:{}",
                                userId, notificationType, notificationDTO.getUserId());

                        // 4. 推送成功后，从ZSet中删除该通知（按value精确匹配）
                        stringRedisTemplate.opsForZSet().remove(
                                offCommentNotification,
                                objectMapper.writeValueAsString(notificationDTO)
                        );

                    } catch (Exception e) {
                        log.error("用户{}的离线{}通知推送失败，通知ID:{}",
                                userId, notificationType, notificationDTO.getUserId(), e);
                        // 可选：推送失败时将通知重新加入ZSet（需考虑重试次数，避免无限循环）
                    }
                });
            }
        }
    public void offMessage(Long userId,WebSocketSession session) {
        //获取有离线消息的房间号
        String offRoomKey=RedisCacheConstants.CHAT_OFFLINE_MESSAGE_LIST_PREFIX+userId;
        Set<Object> keys = stringRedisTemplate.opsForHash().keys(offRoomKey);
        List<Long> roomIds = keys.stream()
                .map(roomIdObj -> Long.parseLong(roomIdObj.toString()))
                .collect(Collectors.toList());

        if(keys==null||keys.isEmpty())
        {
            return;
        }

        //从redis中获取离线消息
        for(Long roomId:roomIds) {
            String offlineListKey = RedisCacheConstants.CHAT_OFFLINE_MESSAGE_LIST_PREFIX + userId +":"+ roomId;
            List<String> messages = stringRedisTemplate.opsForList().range(offlineListKey, 0, -1);
            if (messages == null || messages.isEmpty()) {
                log.info("用户无离线消息");
                return;
            }
            //反序列化消息列表
            List<MessageDTO> messageList = messages.stream()
                    .map(jsonstr -> {
                        try {
                            return objectMapper.readValue(jsonstr, MessageDTO.class);
                        } catch (JsonProcessingException e) {
                            log.error("消息反序列化失败");
                            return null;
                        }
                    })
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            if (messageList.isEmpty()) {
                return;
            }
            final Object sendLock = new Object();

            for (MessageDTO message : messageList) {
                offlineExecutor.submit(() -> {
                    try {
                        if (session!=null&&session.isOpen()) {
                        synchronized (sendLock) {
                            String jsonMessage = objectMapper.writeValueAsString(message);
                            session.sendMessage(new TextMessage(jsonMessage));
                            log.info("离线消息推送成功，userId:{}, messageId:{}", userId, message.getId());
                            //推送成功后，从Redis删除该离线消息
                            stringRedisTemplate.opsForList().remove(offlineListKey, 1,jsonMessage);
                            log.info("删除推送过的离线消息"+jsonMessage);
                        }
                        }else {
                            log.warn("WebSocket连接已关闭，无法推送消息，userId:{}", userId);
                        }
                    } catch (Exception e) {
                        log.error("离线消息推送失败，userId:{}, messageId:{}", userId, message.getId(), e);

                    }
                });
            }
        }
    }
    /**
     *处理消息
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        //发送人的id
        Long userId= (Long) session.getAttributes().get("userId");
        if(message instanceof TextMessage) {
            String text = ((TextMessage) message).getPayload();
            JsonNode jsonNode = objectMapper.readTree(text);
            if(jsonNode.has("type")&&"heartbeat".equals(jsonNode.get("type").asText()))
            {
                String data = jsonNode.path("data").asText();
                if("ping".equals(data)) {
                    log.info("接收到心跳检测，准备回复"+userId);
                    String pongResponse = "{\"type\":\"heartbeat\",\"data\":\"pong\"}";
                    session.sendMessage(new TextMessage(pongResponse));
                    log.info("已回复心跳");
                }
            return;
            }
            ChatMessageDTO chatMessageDTO = objectMapper.readValue(text, ChatMessageDTO.class);
            ChatMessage chatMessage =new ChatMessage();
            chatMessage.setChatRoomId(Long.valueOf(chatMessageDTO.getChatRoomId()));
            chatMessage.setContent(chatMessageDTO.getContent());
            chatMessage.setMessageType(chatMessageDTO.getMessageType());
            chatMessage.setSendTime(new Date());
            chatMessage.setSenderId(userId);
            int count= chatRoomMemberMapper.countMembersByRoomId(chatMessage.getChatRoomId());
            if(count==2)
            {
                pravateChatstrategy.handleMessage(session,chatMessage);
                log.info("发送私信");
            }else if(count>=3)
            {
                groupChatstrategy.handleMessage(session,chatMessage);
            }
        }

    }

    /**
     * 传输层发生错误时触发
     * @param session
     * @param exception
     * @throws Exception
     */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
        // 1. 获取用户ID（从Session属性中）
        Long userId = (Long) session.getAttributes().get("userId");
        if (userId == null) {
            log.error("WebSocket传输错误：无法获取用户ID，Session: {}", session.getId());
            return;
        }

        // 2. 记录错误日志（包含用户ID和异常信息）
        log.error("WebSocket传输错误 - 用户ID: {}, Session: {}, 异常: {}", userId, session.getId(), exception.getMessage(), exception);

    }

    /**
     * 连接关闭后触发
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        Long userId = (Long) session.getAttributes().get("userId");
        if (userId == null) {
            log.warn("连接关闭，无法获取用户ID，Session:{}", session.getId());
            return;
        }
        // 1. 移除会话
        sessionManager.removeSession(userId);
        // 2. 同步更新在线状态为离线（0）
        chatService.updateUserOnlineStatus(userId,0);
       /* String onlineKey = RedisCacheConstants.USER_ONLINE_PREFIX + userId;
        stringRedisTemplate.opsForValue().set(onlineKey, "0", 30,TimeUnit.MINUTES);*/
        log.info("连接关闭，用户离线，userId:{}，status:{}", userId, status);
    }
    }